package com.bw.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.SHiTi.OrderDetail;
import com.bw.SHiTi.OrderInfo;
import com.bw.util.KafkaUtil;
import com.bw.util.MyDIMRequestIO;
import com.sun.xml.internal.bind.v2.TODO;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @BelongsProject: work_new_order
 * @BelongsPackage: com.bw.dws
 * @Author: wuxiaopeng
 * @CreateTime: 2025-07-10  10:34
 * @Description: TODO
 * @Version: 1.0
 */
public class big_table {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//      获取kafka的数据
        DataStreamSource<String> info = KafkaUtil.getKafkaSource(env, "new_work_order_dwd_info", "asdf");
//        info.print();
        DataStreamSource<String> detail = KafkaUtil.getKafkaSource(env, "new_work_order_dwd_detail", "asdf");
//        detail.print();
//        DataStreamSource<String> payment = KafkaUtil.getKafkaSource(env, "new_work_order_dwd_payment", "asdf");
//        DataStreamSource<String> cart = KafkaUtil.getKafkaSource(env, "new_work_order_dwd_cart", "asdf");

//      TODO: 设置 水位线，过滤数据
        SingleOutputStreamOperator<OrderInfo> infoDS = info.map(t -> JSON.parseObject(t, OrderInfo.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getCreateTime()));

        SingleOutputStreamOperator<OrderDetail> detailDS = detail.map(t -> JSON.parseObject(t, OrderDetail.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getCreateTime()));

//      TODO: 合并数据
        SingleOutputStreamOperator<OrderInfo> table_2 = infoDS.keyBy(t -> t.getId())
                .intervalJoin(detailDS.keyBy(t -> t.getOrderId()))
                .between(Time.milliseconds(-2), Time.milliseconds(1))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderInfo>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, ProcessJoinFunction<OrderInfo, OrderDetail, OrderInfo>.Context context, Collector<OrderInfo> collector) throws Exception {
                        orderInfo.setOrderId(orderDetail.getOrderId());
                        orderInfo.setSkuId(orderDetail.getSkuId());
                        orderInfo.setOrderPrice(orderDetail.getOrderPrice());
                        orderInfo.setSkuNum(orderDetail.getSkuNum());
                        orderInfo.setSplitTotalAmount(orderDetail.getSplitTotalAmount());
                        orderInfo.setSplitActivityAmount(orderDetail.getSplitActivityAmount());
                        orderInfo.setSplitCouponAmount(orderDetail.getSplitCouponAmount());

                        collector.collect(orderInfo);
                    }
                });
//        table_2.print();

//      TODO 异步IO
//        关联sku表
        SingleOutputStreamOperator<OrderInfo> table_3 = AsyncDataStream.unorderedWait(table_2, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getSkuId();
            }

            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_sku where id = ?";
            }

            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setSpuId(jsonObject.getLong("spuId"));
                orderInfo.setPrice(jsonObject.getInteger("price"));
                orderInfo.setSkuName(jsonObject.getString("skuName"));
                orderInfo.setTmId(jsonObject.getLong("tmId"));
                orderInfo.setCategory3Id(jsonObject.getLong("category3Id"));
                orderInfo.setIsSale(jsonObject.getInteger("isSale"));

                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);

//        table_3.print();



//       TODO 关联spu表
        SingleOutputStreamOperator<OrderInfo> table_4 = AsyncDataStream.unorderedWait(table_3, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getSpuId();
            }

            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_spu where id = ?";
            }

            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setSpuName(jsonObject.getString("spuName"));
//                System.out.println(jsonObject);
                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);

//        table_4.print();




//      TODO 关联用户表
        SingleOutputStreamOperator<OrderInfo> table_5 = AsyncDataStream.unorderedWait(table_4, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getUserId();
            }

            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_user where id = ?";
            }

            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setUserName(jsonObject.getString("name"));
                orderInfo.setPhoneNum(jsonObject.getString("phoneNum"));
                orderInfo.setBirthday(jsonObject.getString("birthday"));
                orderInfo.setGender(jsonObject.getString("gender"));

//                System.out.println(jsonObject);
                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);

//        table_5.print();

//      TODO 关联c3表
        SingleOutputStreamOperator<OrderInfo> table_6 = AsyncDataStream.unorderedWait(table_5, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getCategory3Id();
            }

            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_c3 where id = ?";
            }

            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setC3name(jsonObject.getString("name"));
                orderInfo.setCategory2Id(jsonObject.getLong("category2Id"));
//                System.out.println(jsonObject);
                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);

//        table_6.print();



//      TODO 关联c2表
        SingleOutputStreamOperator<OrderInfo> table_7 = AsyncDataStream.unorderedWait(table_6, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getCategory2Id();
            }

            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_c2 where id = ?";
            }

            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setC2Name(jsonObject.getString("name"));
                orderInfo.setCategory1Id(jsonObject.getLong("category1Id"));
//                System.out.println(jsonObject);
                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);

//        table_7.print();


//      TODO 关联c1表
        SingleOutputStreamOperator<OrderInfo> table_8 = AsyncDataStream.unorderedWait(table_7, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getCategory1Id();
            }

            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_c1 where id = ?";
            }

            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setC1Name(jsonObject.getString("name"));
//                System.out.println(jsonObject);
                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);


//        table_8.print();




//      TODO 关联province
        SingleOutputStreamOperator<OrderInfo> table_9 = AsyncDataStream.unorderedWait(table_8, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getProvinceId();
            }

            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_province where id = ?";
            }

            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setProvinceName(jsonObject.getString("name"));
                orderInfo.setRegionId(jsonObject.getLong("regionId"));
                orderInfo.setAreaCode(jsonObject.getString("areaCode"));
                orderInfo.setIsoCode(jsonObject.getString("isoCode"));
                orderInfo.setIso31662(jsonObject.getString("iso31662"));
//                System.out.println(jsonObject);
                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);

//        table_9.print();


//      TODO 关联region
        SingleOutputStreamOperator<OrderInfo> table_10 = AsyncDataStream.unorderedWait(table_9, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getRegionId();
            }

            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_region where id = ?";
            }

            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setRegionName(jsonObject.getString("regionName"));
//                System.out.println(jsonObject);
                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);

//        table_10.print();

//      TODO 关联shop
        SingleOutputStreamOperator<OrderInfo> table_11 = AsyncDataStream.unorderedWait(table_10, new MyDIMRequestIO<OrderInfo, OrderInfo>() {
            @Override
            public Object selectKey(OrderInfo orderInfo) {
                return orderInfo.getShop_id();
            }
            @Override
            public String buildSql() {
                return "select * from new_work_order.dim_shop where id = ?";
            }
            @Override
            public OrderInfo join(OrderInfo orderInfo, JSONObject jsonObject) {
                orderInfo.setShopName(jsonObject.getString("shopName"));
                return orderInfo;
            }
        }, 1000, TimeUnit.SECONDS);

        table_11.print();


//     TODO 写入
        table_11.addSink(KafkaUtil.getKafkaSink("big_table"));


        env.execute("work_order");
    }
}
